package com.relytosoft.mizar.analyze.structure.services.consumer;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.relytosoft.dubhe.common.kafka.producer.KafkaOpMsg;
import com.relytosoft.mizar.analyze.base.constant.MediaMqConstants;
import com.relytosoft.mizar.analyze.base.enums.TaskStatusEnum;
import com.relytosoft.mizar.analyze.structure.api.entity.AyStructureTask;
import com.relytosoft.mizar.analyze.structure.api.entity.AyStructureTaskSon;
import com.relytosoft.mizar.analyze.structure.services.mapper.AyStructureTaskSonMapper;
import com.relytosoft.mizar.analyze.structure.services.service.AyStructureTaskService;
import com.relytosoft.mizar.analyze.structure.services.service.AyStructureTaskSonService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.Map;

/**
 * TaskConsumer
 *
 * @Author zyc
 * @Date: 2022/5/18 14:14
 * @Version 1.0
 */
@Component
public class StruTaskConsumer {
    protected Logger log = LoggerFactory.getLogger(this.getClass());

    @Resource
    private AyStructureTaskService ayStructureTaskService;

    @Resource
    private AyStructureTaskSonService ayStructureTaskSonService;

    @Resource
    private AyStructureTaskSonMapper ayStructureTaskSonMapper;

    @KafkaListener(id = "consumer-analyze-stru-task-status", topics = MediaMqConstants.STRU_ANALYZE_TASK_STATUS_TOPIC, containerFactory = "singleContainerFactory")
    public void taskStatusSync(ConsumerRecord<?, ?> record, Acknowledgment ack) {
        try {
            boolean isConsume = false;
            KafkaOpMsg kafkaOpMsg = JSON.parseObject(record.value().toString(), KafkaOpMsg.class);
            String data = kafkaOpMsg.getData();
            //todo 更新子任务状态，需同步更新主任务状态
            if (!StrUtil.isEmptyIfStr(data)) {
                Map<String, Object> map = JSONObject.parseObject(data, new TypeReference<Map<String, Object>>() {
                });
                Integer code = (Integer) map.get("code");
                //子任务id
                String taskId = (String) map.get("taskId");
                //更新任务状态到数据库
                AyStructureTaskSon ayStructureTaskSon = ayStructureTaskSonMapper.getByTaskId(taskId);
                AyStructureTask ayStructureTask = ayStructureTaskService.getById(ayStructureTaskSon.getStructureTaskId());
                if (0 == code) {
                    ayStructureTask.setStatus(TaskStatusEnum.RUNING.getValue());
                    if (TaskStatusEnum.CREATED.getValue() == ayStructureTaskSon.getStatus()) {
                        ayStructureTaskSon.setStatus(TaskStatusEnum.RUNING.getValue());
                    } else {
                        ayStructureTaskSon.setStatus(TaskStatusEnum.FINISHED.getValue());
                    }
                } else {
                    ayStructureTask.setStatus(TaskStatusEnum.UNUSUAL.getValue());
                }
                ayStructureTaskService.updateById(ayStructureTask);
                isConsume = ayStructureTaskSonService.updateById(ayStructureTaskSon);
            }
            log.info("线程:{} 消费-结构化任务状态->{}", Thread.currentThread(), data);
            if (isConsume) {
                ack.acknowledge();
            }
        } catch (Exception e) {
            log.error(e.getMessage());
        }
    }

    @KafkaListener(id = "consumer-analyze-stru-task-id", topics = MediaMqConstants.STRU_ANALYZE_TASK_ID_TOPIC, containerFactory = "singleContainerFactory")
    public void taskIdSync(ConsumerRecord<?, ?> record, Acknowledgment ack) {
        try {
            boolean isConsume = false;
            KafkaOpMsg kafkaOpMsg = JSON.parseObject(record.value().toString(), KafkaOpMsg.class);
            String data = kafkaOpMsg.getData();
            //todo 更新子任务的taskId
            if (!StrUtil.isEmptyIfStr(data)) {
                Map<String, Object> map = JSONObject.parseObject(data, new TypeReference<Map<String, Object>>() {
                });
                //子任务id
                Integer id = (Integer) map.get("id");
                //三方任务id
                String taskId = (String) map.get("taskId");
                //更新三方任务id到数据库
                AyStructureTaskSon ayStructureTaskSon = ayStructureTaskSonService.getById(id);
                ayStructureTaskSon.setTaskId(taskId);
                ayStructureTaskSon.setBeginTime(LocalDateTime.now());
                ayStructureTaskSon.setEndTime(null);
                isConsume = ayStructureTaskSonService.updateById(ayStructureTaskSon);
            }
            log.info("线程:{} 消费-结构化任务taskId关联->{}", Thread.currentThread(), data);
            if (isConsume) {
                ack.acknowledge();
            }
        } catch (Exception e) {
            log.error(e.getMessage());
        }
    }

}
